package cn.edu360

import java.sql.{Connection, DriverManager, ResultSet}

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.rdd.RDD
import scalikejdbc.config.DBs
import scalikejdbc.{DB, SQL}

object Utils {
  def writeToMysql(list: List[(String, Int)]): Unit = {
    val config: Config = ConfigFactory.load()
    list.foreach(it => {
      // 获取mysql的连接
      val url = config.getString("db.url")
      val user = config.getString("db.user")
      val password = config.getString("db.password")
      var conn: Connection = null
      try {
        // 获取到一个操作mysql的连接
        conn = DriverManager.getConnection(url, user, password)

        // 开启事务
        conn.setAutoCommit(false)

        val table = config.getString("db.table")
        // 创建一个数据库
        val createPstm = conn.prepareStatement(s"create table if not exists ${table} (proAndcity varchar(20),cnts int)")
        createPstm.execute()
        createPstm.close()

        // tp 是要操作的最终的数据

        // 1，先去mysql数据库中查询 是否存在数据   // 添加一个update
        val queryPstm = conn.prepareStatement(s"select cnts from ${table} where proAndcity = ? for update")
        queryPstm.setString(1, it._1)
        // 得到查询的结果
        val result: ResultSet = queryPstm.executeQuery()

        // 定义一个标识 用于判断是否需要直接入库
        var needInsert = true
        while (result.next()) {

          needInsert = false
          // 2 存在，就取出数据   累加   update
          val oldValues = result.getInt(1)
          val newValues = oldValues + it._2

          val upPstm = conn.prepareStatement(s"update ${table} set cnts = ?  where proAndcity = ? ")
          upPstm.setInt(1, newValues)
          upPstm.setString(2, it._1)
          // 执行 更新
          upPstm.executeUpdate()

          if (upPstm != null) upPstm.close()
        }

        // 3 如果不存在，直接插入数据
        if (needInsert) {
          // 执行插入数据
          val insertPstm = conn.prepareStatement(s"insert into ${table} values (?,?)")

          insertPstm.setString(1, it._1)
          insertPstm.setInt(2, it._2)
          insertPstm.execute()
          if (insertPstm != null) insertPstm.close()
        }

        // 手动提交
        conn.commit()
      } catch {
        case e: Exception => e.printStackTrace()
      }
      finally {
        // 释放资源
        if (conn != null) conn.close()
      }
    })
  }

  //scalikeJDbc版本
  def scalijeJdbcToMysql(generalResult: RDD[(String, Int)]): Unit = {
    // 加载数据库的配置信息
    DBs.setup()
    // 解析application.conf文件
    // val config: Config = ConfigFactory.load()
    generalResult.foreachPartition(iter => {
      // 存在即更新，不存在插入
      DB.localTx { implicit session =>
        iter.foreach(tp => {
          SQL(
            """
              |INSERT INTO t_reportAnother (proAndcity, cnts)
              |VALUES (?, ?) ON DUPLICATE KEY
              |UPDATE proAndcity = proAndcity + ?,cnts = cnts + ?
            """.stripMargin).bind(tp._1, tp._2, tp._1, tp._2
          ).update().apply()
        })
      }
    })
  }
}